package cn.gwm.flink.streaming.task;

import cn.gwm.flink.streaming.beans.BeanSource;
import cn.gwm.flink.streaming.constant.BaseFields;
import cn.gwm.flink.streaming.constant.VehicleType;
import cn.gwm.flink.streaming.factory.DataHubSourceFactory;
import cn.gwm.flink.streaming.ods.OdsPreTask;
import cn.gwm.flink.streaming.ods.model.ModelCanEnum;
import cn.gwm.flink.streaming.ods.model.ModelGpsEnum;
import cn.gwm.flink.streaming.sink.kafka.FlinkKafkaUtil;
import cn.gwm.utils.ConfigLoader;
import cn.hutool.core.date.DatePattern;
import cn.hutool.core.date.DateUtil;
import cn.hutool.json.JSONObject;
import cn.hutool.json.JSONUtil;
import com.aliyun.datahub.client.model.RecordEntry;
import com.aliyun.datahub.client.model.TupleRecordData;
import lombok.extern.slf4j.Slf4j;
import org.apache.flink.api.common.eventtime.*;
import org.apache.flink.api.common.functions.CoGroupFunction;
import org.apache.flink.api.common.functions.FlatMapFunction;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.common.restartstrategy.RestartStrategies;
import org.apache.flink.api.common.state.*;
import org.apache.flink.api.common.time.Time;
import org.apache.flink.api.common.typeinfo.Types;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.streaming.api.datastream.BroadcastStream;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.ProcessFunction;
import org.apache.flink.streaming.api.functions.co.CoProcessFunction;
import org.apache.flink.streaming.api.functions.co.KeyedBroadcastProcessFunction;
import org.apache.flink.streaming.api.functions.source.SourceFunction;
import org.apache.flink.streaming.connectors.kafka.FlinkKafkaConsumer;
import org.apache.flink.util.Collector;
import org.apache.flink.util.OutputTag;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.configuration.RestOptions;

import java.time.Duration;
import java.util.HashMap;
import java.util.Iterator;
import java.util.Map;
import java.util.Random;

/**
 * @ClassName OdesDemoTask
 * @Author xzh
 * @Date 2022/11/28 17:11
 **/
@Slf4j
public class OdsDemoTask {

    public static class Action{
        public String userId;
        public String action;

        public Action(String userId, String action) {
            this.userId = userId;
            this.action = action;
        }

        @Override
        public String toString() {
            return "Action{" +
                    "userId='" + userId + '\'' +
                    ", action='" + action + '\'' +
                    '}';
        }
    }

    public static class Pattern{
        public String action1;
        public String action2;

        public Pattern(String action1, String action2) {
            this.action1 = action1;
            this.action2 = action2;
        }

        @Override
        public String toString() {
            return "Pattern{" +
                    "action1='" + action1 + '\'' +
                    ", action2='" + action2 + '\'' +
                    '}';
        }
    }

    public static void main(String[] args) throws Exception {
        ConfigLoader.init(args);
        Configuration configuration = new Configuration();
        configuration.setInteger(RestOptions.PORT, 8081);
        final StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment(configuration);
        env.setParallelism(1);
        // 读取用户行为事件流
        DataStreamSource<Action> actionStream = env.fromElements(
                new Action("Alice", "login"),
                new Action("Alice", "pay"),
                new Action("Bob", "login"),
                new Action("Bob", "buy"));
        // 定义行为模式流，代表了要检测的标准
        DataStreamSource<Pattern> patternStream = env.fromElements(
                new Pattern("login", "buy"),
                new Pattern("login", "pay"));
        // 定义广播状态的描述 器，创建广播流
        MapStateDescriptor<String, Pattern> bcStateDescriptor = new MapStateDescriptor<>("patterns",String.class, Pattern.class);
        BroadcastStream<Pattern> bcPatterns = patternStream.broadcast(bcStateDescriptor);
        // 将事件流和广播流连接起来，进行处理
        DataStream<Tuple2<String, Pattern>> matches = actionStream
                .keyBy(data -> data.userId)
                .connect(bcPatterns)
                .process(new PatternEvaluator());
        matches.print();
        env.setRestartStrategy(RestartStrategies.fixedDelayRestart(
                3,
                Time.seconds(10)
        ));
        env.execute("OdsDemoTask");
    }

    public static class PatternEvaluator extends KeyedBroadcastProcessFunction<String,Action,Pattern,Tuple2<String, Pattern>> {
        // 定义一个值状态，保存上一次用户行为
        ValueState<String> prevActionState;

        @Override
        public void open(Configuration parameters) throws Exception {
            prevActionState = getRuntimeContext().getState(new ValueStateDescriptor<>("lastAction", String.class));
        }

        @Override
        public void processElement(Action action, ReadOnlyContext ctx, Collector<Tuple2<String, Pattern>> out) throws Exception {
            Pattern pattern = ctx.getBroadcastState(new MapStateDescriptor<>("patterns",String.class, Pattern.class)).get(action.action);
            String prevAction = prevActionState.value();
            if (pattern != null && prevAction != null) {
                // 如果前后两次行为都符合模式定义，输出一组匹配
                if (pattern.action1.equals(prevAction) && pattern.action2.equals(action.action)) {
                    out.collect(new Tuple2<>(ctx.getCurrentKey(), pattern));
                }
            }
            prevActionState.update(action.action);
        }

        @Override
        public void processBroadcastElement(Pattern pattern, Context ctx, Collector<Tuple2<String, Pattern>> out) throws Exception {
            BroadcastState<String, Pattern> bcState = ctx.getBroadcastState(new MapStateDescriptor<>("patterns", String.class, Pattern.class));
            // 将广播状态更新为当前的 pattern
            bcState.put(pattern.action2, pattern);
        }
    }

    public static void main2(String[] args) throws Exception {
        ConfigLoader.init(args);
        final StreamExecutionEnvironment env =  StreamExecutionEnvironment.getExecutionEnvironment();

        Map<String, OutputTag<String>> tagMap = new HashMap<>();
        for (ModelCanEnum canEnum : ModelCanEnum.values()) {
            tagMap.put(canEnum.getModelName(), new OutputTag<String>(canEnum.getValue()) {
            });
        }
        SingleOutputStreamOperator<String> process = OdsPreTask.defindStream(env, tagMap,
                "1673508274399TD6XY", "2023-03-05 05:00:00");
        SingleOutputStreamOperator<JSONObject> map = process.getSideOutput(tagMap.get(ModelCanEnum.ES24.getModelName()))
                .map(JSONUtil::parseObj).map(item -> {
            System.out.println(item.size());
            return item;
        });
        map.print();
        env.setParallelism(1);
        env.execute("OdsDemoTask");
    }

    public static void main1(String[] args) throws Exception {
        ConfigLoader.init(args);
        final StreamExecutionEnvironment env =  StreamExecutionEnvironment.getExecutionEnvironment();
        // aliyun test
        FlinkKafkaConsumer<String> kafkaConsumer = FlinkKafkaUtil.getConsumer(
                "10.31.8.112:9092,10.31.8.113:9092,10.31.8.114:9092",
                "consumer-ods-can-es13-test",
                "ods_can_es13"
        );
        kafkaConsumer.setStartFromEarliest();
        DataStreamSource<String> stringDataStreamSource = env.addSource(kafkaConsumer);
        stringDataStreamSource.print();
        env.setParallelism(1);
        env.execute("OdsDemoTask");
    }

    public static void main3(String[] args) throws Exception {
        Configuration configuration = new Configuration();
        configuration.setInteger(RestOptions.PORT, 8081);
        final StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment(configuration);
        env.setParallelism(1);
        DataStreamSource<String> stream1 = env.socketTextStream("bd-zcpt-bd-zcpt-test-000001", 7777);
        DataStreamSource<String> stream3 = env.socketTextStream("bd-zcpt-bd-zcpt-test-000003", 7777);
//        long sid= System.currentTimeMillis()/1000*1000-24*60*60*1000L;
//        DataStreamSource<String> stream1 =  env.addSource(new ClickSource("state1",sid));
//        DataStreamSource<String> stream3 = env.addSource(new ClickSource("state2",sid));
//        SingleOutputStreamOperator<JSONObject> flatMap1 = filterData(stream1).name("0001");
//        SingleOutputStreamOperator<JSONObject> flatMap3 = filterData(stream3).name("0002");
//        SingleOutputStreamOperator<JSONObject> process = flatMap1.connect(flatMap3).keyBy(data -> data.getStr("vin"), data -> data.getStr("vin")).process(new ConnectProcess());
//        flatMap1.print("flatMap1");
//        flatMap3.print("flatMap3");
//        process.print("process");

        SingleOutputStreamOperator<Event> s1 = stream1.map(new MapFunction<String, Event>() {
            @Override
            public Event map(String s) throws Exception {
                String[] split = s.split(",");
                return new Event(split[0], split[1], Long.valueOf(split[2]));
            }
        }).assignTimestampsAndWatermarks(WatermarkStrategy.<Event>forBoundedOutOfOrderness(Duration.ofSeconds(2))
                .withTimestampAssigner((SerializableTimestampAssigner<Event>) (event, l) -> event.time));
        SingleOutputStreamOperator<Event> s3 = stream3.map(new MapFunction<String, Event>() {
            @Override
            public Event map(String s) throws Exception {
                String[] split = s.split(",");
                return new Event(split[0], split[1], Long.valueOf(split[2]));
            }
        }).assignTimestampsAndWatermarks(WatermarkStrategy.<Event>forBoundedOutOfOrderness(Duration.ofSeconds(2))
                .withTimestampAssigner((SerializableTimestampAssigner<Event>) (event, l) -> event.time));
        s1.print("s1");
        s3.print("s3");
        SingleOutputStreamOperator<String> process = s1.union(s3).process(new ProcessFunction<Event, String>() {
            @Override
            public void processElement(Event value, Context ctx, Collector<String> out) throws Exception {
                System.out.println(value.toString());
                out.collect(" 水位线： ：" + ctx.timerService().currentWatermark());
            }
        });
        process.print();
        env.setRestartStrategy(RestartStrategies.fixedDelayRestart(
                3,
                Time.seconds(10)
        ));
        env.execute("testWater");
    }

    private static SingleOutputStreamOperator<JSONObject> filterData(DataStreamSource<String> stream) {
        return stream.flatMap(new FlatMapFunction<String, JSONObject>() {
            @Override
            public void flatMap(String s, Collector<JSONObject> collector) throws Exception {
                try {
                    JSONObject json = JSONUtil.parseObj(s);
                    if (json != null && json.getLong("time") != null && json.getStr("vin")!=null) {
                        collector.collect(json);
                    } else {
                        log.error("数据中缺少必要有效字段，time.数据：", s);
                    }
                } catch (Exception e) {
                    log.error("流中得到数据不是json字符传，而是：{}", s);
                }
            }
        }).assignTimestampsAndWatermarks(WatermarkStrategy.<JSONObject>forMonotonousTimestamps()
                .withTimestampAssigner((element, recordTimestamp) -> element.getLong("time")));
    }
    static class Event{
        private String name;
        private String url;
        private long time;
        public Event(String name,String url,long time){
            this.name = name;
            this.url = url;
            this.time = time;
        }

        @Override
        public String toString() {
            return "Event{" +
                    "name='" + name + '\'' +
                    ", url='" + url + '\'' +
                    ", time=" + time +
                    '}';
        }
    }

    @Slf4j
    static class ConnectProcess extends CoProcessFunction<JSONObject, JSONObject, JSONObject> {

        private transient ValueState<JSONObject> state1;
        private transient ValueState<JSONObject> state2;
        private transient MapState<String, JSONObject> mapState;
        private final long DELAY_TIME = 60*60*1000L;


        @Override
        public void open(Configuration parameters) throws Exception {
            state1 = getRuntimeContext().getState(new ValueStateDescriptor<>("state1",JSONObject.class));
            state2 = getRuntimeContext().getState(new ValueStateDescriptor<>("state2",JSONObject.class));
            MapStateDescriptor<String, JSONObject> mapState = new MapStateDescriptor<>("mapState", String.class, JSONObject.class);
            this.mapState = getRuntimeContext().getMapState(mapState);
        }

        @Override
        public void onTimer(long timestamp, OnTimerContext ctx, Collector<JSONObject> out) throws Exception {
            log.info("onTimer,timestamp={}",timestamp);
            if(state1!=null&&state1.value()!=null){
                JSONObject object = state1.value();
                if(object!=null){
                    if(!mapState.isEmpty()){
                        if(mapState.contains(object.getStr("time"))){
                            JSONObject obj = mapState.get(object.getStr("time"));
                            if(obj!=null) {
                               object.putAll(obj);
                            }
                        }
                        Iterator<Map.Entry<String, JSONObject>> iterator = mapState.iterator();
                        while (iterator.hasNext()){
                            Map.Entry<String, JSONObject> next = iterator.next();
                            //当can数据延迟落后gps10分钟后，放弃合并
                            if( next!=null && next.getKey().compareTo(object.getStr("time"))<=0){
                                iterator.remove();
                            }
                        }
                    }
                    out.collect(object);
                    state1.clear();
                }
            }
        }

        @Override
        public void processElement1(JSONObject object, Context context, Collector<JSONObject> collector) throws Exception {
            if(!mapState.isEmpty()){
                if(mapState.contains(object.getStr("time"))){
                    JSONObject obj = mapState.get(object.getStr("time"));
                    if(obj!=null){
                        object.putAll(obj);
                        collector.collect(object);
                    }else{
                        state1.update(object);
                        context.timerService().registerEventTimeTimer(object.getLong("time") + 200L);
                    }
                }else{
                    state1.update(object);
                    context.timerService().registerEventTimeTimer(object.getLong("time") + 200L);
                }
                Iterator<Map.Entry<String, JSONObject>> iterator = mapState.iterator();
                while (iterator.hasNext()){
                    Map.Entry<String, JSONObject> next = iterator.next();
                    if( next!=null && next.getKey().compareTo(object.getStr("time"))<=0){
                        iterator.remove();
                    }
                }

            }else{
                state1.update(object);
                context.timerService().registerEventTimeTimer(object.getLong("time") + 2000L);
            }
        }

        @Override
        public void processElement2(JSONObject obj, Context context, Collector<JSONObject> collector) throws Exception {
            if(state1!=null&&state1.value()!=null){
                JSONObject object = state1.value();
                if(object!=null){
                    if(object.getStr("time").equals(obj.getStr("time"))){
                        object.putAll(obj);
                        collector.collect(object);
                        state1.clear();
                    }else {
                        mapState.put(obj.getStr("time"), obj);
                    }
                }else{
                    mapState.put(obj.getStr("time"), obj);
                }
            }else{
                mapState.put(obj.getStr("time"), obj);
            }
        }
    }
    static class ClickSource implements SourceFunction<String>{

        private volatile Boolean running = true;
        private long sid;
        private String vin;
        public ClickSource(String vin,long sid){
            this.vin = vin;
            this.sid = sid;
        }

        @Override
        public void run(SourceContext<String> ctx) throws Exception {
            Random random = new Random();
            long tid = 1000;
            while (running){
                JSONObject object = JSONUtil.createObj();
                object.set("vin","aa");

                if("state1".equals(this.vin)){
                    object.set("random",random.nextInt(10));
                    object.set("nm1", this.vin);
                }
                if("state2".equals(this.vin)) {
                    object.set("aid", System.currentTimeMillis());
                    object.set("nm2", this.vin);
                }


                JSONObject object1 = JSONUtil.createObj();
                object1.set("vin","bb");
                if(random.nextInt(10)>7){
                    object1.set("time",sid-tid);
                    object.set("time",sid-tid);
                }else{
                    object.set("time",sid+tid);
                    object1.set("time",sid+tid);

                }
                if("state1".equals(this.vin)){
                    object1.set("random",random.nextInt(10));
                    object1.set("nm1", this.vin);
                }
                if("state2".equals(this.vin)) {
                    object1.set("aid", System.currentTimeMillis());
                    object1.set("nm2", this.vin);
                }
                ctx.collect(object.toString());
                ctx.collect(object1.toString());
                Thread.sleep(10000L);
                tid += 1000L;
            }
        }

        @Override
        public void cancel() {
            running = false;
        }
    }
}
